package org.apache.rocketmq.client.codecoord.consumer;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.codecoord.RocketMqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * pull模式消息发送
 *
 * @author tianxincoord@163.com
 * @since 2021/10/25
 */
public class MessagePullConsumer {
    private static final Map<MessageQueue, Long> OFFSET_TABLE = new HashMap<>();

    public static void main(String[] args) throws Exception {
        MessagePullConsumer messageSend = new MessagePullConsumer();
        messageSend.pullConsumer();
    }

    public void pullConsumer() throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer();
        consumer.setConsumerGroup(RocketMqConstant.GROUP);
        consumer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDRESS);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.start();

        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(RocketMqConstant.TOPIC);
        for (MessageQueue messageQueue : messageQueues) {
            long consumeOffset = consumer.fetchConsumeOffset(messageQueue, true);
            System.out.printf("【%s】offset【%s】%n", JSONObject.toJSON(messageQueue), consumeOffset);

            SINGLE_ME:
            while (true) {
                long offset = getMessageQueueOffset(messageQueue);
                // 如果没有消息，将会阻塞
                PullResult pullResult = consumer.pullBlockIfNotFound(messageQueue, null, offset, 1);
                putMessageQueueOffset(messageQueue, pullResult.getNextBeginOffset());
                switch (pullResult.getPullStatus()) {
                    case NO_NEW_MSG:
                    case NO_MATCHED_MSG:
                        System.out.println("未获取新消息");
                        break SINGLE_ME;
                    case OFFSET_ILLEGAL:
                        System.out.println("偏移量不合法");
                        break;
                    case FOUND:
                        System.out.printf("拉取结果【%s】%n", new String(pullResult.getMsgFoundList().get(0).getBody()));
                        break;
                    default:
                        System.out.println("未知错误，程序退出：" + pullResult.getPullStatus());
                        return;
                }
            }
        }
        consumer.shutdown();
    }

    private static long getMessageQueueOffset(MessageQueue messageQueue) {
        return OFFSET_TABLE.getOrDefault(messageQueue, 0L);
    }

    private static void putMessageQueueOffset(MessageQueue messageQueue, long offset) {
        OFFSET_TABLE.put(messageQueue, offset);
    }
}
